//
// JSDownloader_New.cpp : Defines the entry point for the console application.
// code style: TAB is forbidden, the indent is 2 whitespace in this file
//

#if defined(_MSC_VER)
// Specifies that the minimum required platform is Windows XP.
#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0501
#endif

#define _CRT_SECURE_NO_WARNINGS
#endif

#include <stdio.h>
#include <malloc.h>
#include <string.h>
#include <iostream>
#include <sstream>
#include <sys/types.h>
#include <uuid/uuid.h>
#include <errno.h>
#include <iconv.h>
#include <signal.h>
#include <sys/wait.h>
#include "v8/v8.h"
#include "circlequeue_shm.h"
#include "block_queue.h"
#include "thread_pool.h"

#define CURL_STATICLIB
#include "curl/curl.h"
#include <sys/stat.h>

#include "rapidxml-1.13/rapidxml.hpp"
#include "rapidxml-1.13/rapidxml_utils.hpp"
#include "rapidxml-1.13/rapidxml_print.hpp"

#include "hash.h"
#include "mem_pool.h"

using namespace std;
using namespace v8;
using namespace rapidxml;

#define SEP_INDICATOR_CHAR ','
#define SEP_INDICATOR_STR ","
#define XML_ROOT_STR "/"
#define QUEUE_MOREDATA_PATH "/var/more_entity"

#define PAGE_CACHE_SIZE 512*1024
#define MAX_REDIRECT_NUM  256
#define MAX_PATH_LEN 1024

#define ENTITY_MAX_URL_BUFF  1024
#define ENTITY_MAX_POST_DATA_BUFF  4096
#define ENTITY_MAX_OUTPATH_BUFF  1024
#define ENTITY_MAX_DOMAIN_BUFF  30
#define JS_MAX_PATH_BUFF  1024
#define NOTIFY_UTCC_URL   1024

#define USE_CURL_API_DOWNLOAD
#define MAX_DISPATHER_THREADS  8

//#define NOTIFY_VSYS_URL "http://118.186.212.98/videoUploadApi.htm"
#define NOTIFY_SUCCESS_VSYS_URL "http://122.11.45.110/directApi.htm"
#define NOTIFY_ZHUANTIE_FAILURE_URL "http://zhuantie.ku6.com/index.php?a=fail&c=index&%s"
#define NOTIFY_ZHUANTIE_SUCCESS_URL "http://zhuantie.ku6.com/index.php?a=success&c=index&%s"

#define NOTIFY_ZHUAQU_URL "http://zhuaqu.ku6.com/task/notify.do?state=%d&%s"
//#define NOTIFY_ZHUAQU_FAILURE_URL "http://zhuaqu.ku6.com/task/notify.do?state=-1&%s"

#define FAILURE_NOTIFY  -1
#define SUCCESS_NOTIFY  0

#define OVERDURA_FAILURE_NOTIFY -2
#define ZHUAQU_SUCCESS_NOTIFY    2
#define ZHUAQU_FAILURE_NOTIFY   -1

#define DOWNLOAD_PAGE_TIMEOUT  60
#define DOWNLOAD_FILE_TIMEOUT  60*10*100
#define DOWNLOAD_MMS_FILE_TIMEOUT  60*10
#define NOTIFY_UTCC_TIMEOUT  60*5
#define MERGE_FILE_TIMEOUT   60*20

#define INPUT_FILE_ARG_FORMAT " -i "
#define OUTPUT_FILE_ARG_FORMAT " -o "

typedef struct WebPage {
  char *data;
  size_t size;
  size_t left;
}WebPage;

typedef struct {
   queue_t queue;
   char    data[ENTITY_MAX_POST_DATA_BUFF+1];
   char    url[ENTITY_MAX_URL_BUFF+1];
   char    path[ENTITY_MAX_OUTPATH_BUFF+1];
   char    finalname[ENTITY_MAX_OUTPATH_BUFF+1];
   char    domain[ENTITY_MAX_DOMAIN_BUFF+1];
} entity_t;

//block queue for store free urls
block_queue_t free_entities;

//block queue for store urls and  worker thread to receive url
block_queue_t worker_entities;

hash_t *JSData;

pthread_rwlock_t JSData_RWlock;

pthread_t **threads;

char input_js[JS_MAX_PATH_BUFF+1]             = {'\0'};

char notify_utcc_url[NOTIFY_UTCC_URL+1]       = {'\0'};

volatile unsigned int  RunCheckFileThread = 1;

#if defined(_MSC_VER)
#pragma comment(lib,"Winmm.lib")
#pragma comment(lib,"Ws2_32.lib")
#pragma comment(lib,"v8_base.lib")
#pragma comment(lib,"v8_snapshot.lib")

#pragma comment(lib,"libmingwex.a")
#pragma comment(lib,"libgcc.a")
#pragma comment(lib,"libwldap32.a")
#pragma comment(lib,"libssl.a")
#pragma comment(lib,"libcrypto.a")
#pragma comment(lib,"librtmp.a")
#pragma comment(lib,"libz.a")
#pragma comment(lib,"libcurl.a")

// MAGIC code for the fuck-needed linker of Visual C++
// just a stub to force CRT to link _fstat64,...
// MUST not be static function
int fuck_fstati64( int fd, struct _stat64 *buffer)
{
  // Note: we can not use _fstati64() because it is a marcro of _fstat64()
  return _fstat64(fd, buffer);
}

int fuck_stati64(const char * name, struct _stat64 * stat)
{
  return _stat64(name, stat);
}
#endif

//TODO: use smart_ptr

// The downloader server MUST be deployed on Dual-line machine,
// otherwise, some dispatchers will deliver wrong destination media based on DNS.

// we use IE's user-agent, some websites(such as http://v.sohu.com) refuse 3rd user-agent
#if 0
static const char g_user_agent[]={"curl/7.21.7 (i686-pc-mingw32) libcurl/7.21.7 OpenSSL/1.0.0d zlib/1.2.5 librtmp/2.3"};
#else
static const char g_user_agent[]={"Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; .NET CLR 3.0.04506.648; .NET CLR 3.5.21022; .NET CLR 3.0.4506.2152; .NET CLR 3.5.30729; .NET CLR 1.1.4322; .NET4.0C; .NET4.0E)"};
#endif

typedef size_t (*CURL_WRITE_CALLBACK)(const void *, size_t, size_t, void *);

void  destoryAll();

int code_convert(char *f,char *t,char *inbuf,size_t inlen,char *outbuf,size_t outlen)
{
   if(!f || !t || !inbuf) return -1;
   iconv_t cd;
   char **pin  = &inbuf;
   char **pout = &outbuf;

   cd = iconv_open(t,f);
   if (cd==0)
     return -1;

   if (iconv(cd,pin,&inlen,pout,&outlen) == (size_t)-1)
     return -1;
   iconv_close(cd);
   return 0;
}

static size_t curl_write_mem(const void *ptr, size_t size, size_t count, void *stream)
{
  WebPage *page = (WebPage*)stream;

  size *= count;
  if (!page || size <= 0)
    return 0;
  // +1 for the termination: NULL
  if (page->left < size+1) {
    printf("thread_realloc() left:%ld size:%ld  newsize:%ld page->data:%p\n"
           ,page->left,page->size,page->size+page->left+size+PAGE_CACHE_SIZE,page->data);
    char *newbuf = (char*)thread_realloc(page->data, page->size + page->left, page->size+page->left+size+PAGE_CACHE_SIZE);
    if (!newbuf)
      return 0;
    page->data = newbuf;
    page->left += size+PAGE_CACHE_SIZE;
  }
  memcpy(page->data + page->size, ptr, size);
  page->left -= size;
  page->size += size;
  // Terminated with NULL
  page->data[page->size] = 0;

  return size;
}

static size_t curl_write_file(const void *ptr, size_t size, size_t count, void *stream)
{
  if (size <= 0 || !stream){
    printf("curl_write_file() size:%ld\n",size);
    return 0;
  }
  fwrite(ptr, size, count, (FILE*)stream);

  return size * count;
}

static int curl_download(CURL_WRITE_CALLBACK callback, void *opaque, char *url, int method, void *param, int progress,int timeout)
{
  char error_buf[CURL_ERROR_SIZE+1];
  CURL *curl = curl_easy_init();

  if (!curl)
    return -1;

#ifdef _DEBUG
  curl_easy_setopt(curl,CURLOPT_VERBOSE, 1);
#endif

  curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);

  curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, error_buf);
  curl_easy_setopt(curl, CURLOPT_URL, url);
  // Data receiver
  curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, callback);
  curl_easy_setopt(curl, CURLOPT_WRITEDATA, opaque);
  if (method) {
    curl_easy_setopt(curl, CURLOPT_POST, 1);
    curl_easy_setopt(curl, CURLOPT_POSTFIELDS, param);
  }
  // We emulate the curl App's default progress msg
  if (progress) {
    curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0);
    curl_easy_setopt(curl, CURLOPT_STDERR, NULL);
  }

  if(timeout){
    curl_easy_setopt(curl, CURLOPT_TIMEOUT, timeout);
    curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT,DOWNLOAD_PAGE_TIMEOUT);
  }

  // Avoid dead-loop: we limit the MAX redir num to |MAX_REDIRECT_NUM|
  curl_easy_setopt(curl, CURLOPT_MAXREDIRS, MAX_REDIRECT_NUM);
  // MUST set this option to follow 302, 301
  curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1);
  // Explicitly announce we support "deflate, gzip"
  curl_easy_setopt(curl, CURLOPT_ACCEPT_ENCODING, "");
  // MUST setup user-agent as same as curl standlone application,
  // otherwise, some web servers(eg: http://www.tudou.com)
  // will refuse curl application's HTTP request
  curl_easy_setopt(curl, CURLOPT_USERAGENT, g_user_agent);
  CURLcode ret = curl_easy_perform(curl);
  if (ret != CURLE_OK) {
    printf("curl_easy_perform():result(%d, %s) url=%s\n",ret, curl_easy_strerror(ret), url);
    curl_easy_cleanup(curl);
    return -1;
  }
  curl_easy_cleanup(curl);

  return 0;
}

static Handle<Value> ku6_download_page(const Arguments &args)
{
  HandleScope handle_scope;
  if (args.Length() >= 1 && args[0]->IsString()) {
    WebPage page;
    String::AsciiValue url(args[0]->ToString());

    memset(&page, 0, sizeof(WebPage));
    if (curl_download(curl_write_mem, &page, *url, 0, NULL, 0,DOWNLOAD_PAGE_TIMEOUT) < 0) {
      thread_free(page.data,page.size+page.left);
      return Undefined();
    }
    Handle<String> hString;
    if(args.Length() >= 2 && args[1]->IsString()){
       String::AsciiValue f(args[1]->ToString());
       char  t[16] =  {"UTF8"};
       if(args.Length() == 3){
         String::AsciiValue t2(args[2]->ToString());
         strcpy(t,*t2);
       }

       int inlen  = page.size;
       int outlen = inlen;
       if(strcmp(*f,"GBK") == 0 || strcmp(*f,"gbk") == 0)
         outlen = inlen * 4;
       printf("ku6_download_page():the page size:%d\n",outlen);
       char *outbuf = (char*)thread_calloc(outlen+1);
       if(!outbuf){
         thread_free(page.data,page.size+page.left);
         return Undefined();
       }

       int ret = code_convert(*f,t,page.data,inlen,outbuf,outlen);
       printf("ku6_download_page():code_convert result:%d\n",ret);
       if(ret == 0)
         hString = String::New(outbuf,outlen);
       else
         hString = String::New(page.data,page.size);
       thread_free(outbuf,outlen+1);
    }else{
       hString = String::New(page.data, page.size);
    }
    thread_free(page.data,page.size+page.left);
    return handle_scope.Close(hString);
  }

  return Undefined();
}

static Handle<Value> ku6_download_page_post(const Arguments &args)
{
  HandleScope handle_scope;

  if (args.Length() == 2 && args[0]->IsString()) {
    WebPage page;
    String::AsciiValue url(args[0]->ToString());
    String::AsciiValue param(args[1]->ToString());

    memset(&page, 0, sizeof(WebPage));
    if (curl_download(curl_write_mem, &page, *url, 1, *param, 0,DOWNLOAD_PAGE_TIMEOUT) < 0) {
      thread_free(page.data,page.size+page.left);
      return Undefined();
    }
    Handle<String> hString = String::New(page.data, page.size);
    thread_free(page.data,page.size+page.left);
    return handle_scope.Close(hString);
  }

  return Undefined();
}

static int rapidxml_parse_xml(const Arguments &args, int argc, int bAttribute, ostringstream &os)
{
  int i, bFind = 0;

  // ----------------------------------------------
  // Note: MUST use UTF-8 String, otherwise,
  // the internal V8 will encode the string to UTF-8 forcely
  //-----------------------------------------------
  String::Utf8Value xml(args[0]->ToString());
  xml_document<>  doc;
  doc.parse<0>(*xml);
  xml_node<>* root = doc.first_node();
  if (!root)
    return -1;

  argc--;
  String::AsciiValue **pKeyStub = new String::AsciiValue*[argc];
  if (!pKeyStub)
    return -1;
  memset(pKeyStub, 0, sizeof(String::AsciiValue *) * argc);

  for (i= 0; i<argc; i++) {
    pKeyStub[i] = new String::AsciiValue(args[i+1]->ToString());
    if (!pKeyStub[i])
      break;
  }

  if (i != argc) {
    for (i=0; i<argc; i++)
      delete pKeyStub[i];
    delete pKeyStub;
    return -1;
  }

  int argc_key = argc;
  // The last argument is |xml_attribute|
  String::AsciiValue attribute_key(args[argc]->ToString());
  if (bAttribute)
    argc_key--;//We skip the last key which is |xml_attribute|

  xml_node<>* node = root->first_node(**pKeyStub[0]);
  for (; node; node = node->next_sibling(**pKeyStub[0])) {
    xml_node<>* sub_node = node;
    for (i=1; i<argc_key; i++) {
      sub_node = sub_node->first_node(**pKeyStub[i]);
      if (!sub_node)
        break;
    }

    if (bFind)
      os << SEP_INDICATOR_STR;
    if (i >= argc_key && sub_node) {
      if (bAttribute) {
        xml_attribute<> *attr_node = sub_node->first_attribute();
        for (; attr_node; attr_node = attr_node->next_attribute()) {
          char *pAttribute = attr_node->name();
          if (pAttribute && !strcmp(pAttribute, *attribute_key)) {
            os << attr_node->value();
            bFind=1;
            break;
          }
        }
      } else {
        xml_node<>* child = sub_node->first_node();
        if (child)
          os << child->value();
        else
          os << sub_node->value();
        bFind=1;
      }
    }
  }

  // Give the root a chance
  if (!bFind) {
    String::AsciiValue root_str(args[1]->ToString());
    if (*root_str && !strcmp(*root_str, XML_ROOT_STR)) {
      if (bAttribute) {
        xml_attribute<> *attr_node = root->first_attribute();
        for (; attr_node; attr_node = attr_node->next_attribute()) {
          char *pAttribute = attr_node->name();
          if (pAttribute && !strcmp(pAttribute, *attribute_key)) {
            os << attr_node->value();
            bFind=1;
            break;
          }
        }
      } else {
        xml_node<>* child = root->first_node();
        if (child)
          os << child->value();
        else
          os << root->value();
        bFind=1;
      }
    }
  }

  for (i=0; i<argc; i++)
    delete pKeyStub[i];
  delete pKeyStub;

  return 0;
}

static Handle<Value> ku6_parse_xml(const Arguments &args)
{
  HandleScope handle_scope;
  int argc = args.Length();

  if (argc < 2)
    return Undefined();

  ostringstream os;
  try {
    if (rapidxml_parse_xml(args, argc, 0, os) < 0)
      return Undefined();
  } catch (parse_error e) {
    printf("rapidxml_parse_xml() : %s when parse the data at %s\n", e.what(), e.where<char>());
    return String::New("",0);
  }
  Handle<String> hString = String::New(os.str().c_str(), os.str().size());

  return handle_scope.Close(hString);
}

static Handle<Value> ku6_parse_xml_attribute(const Arguments &args)
{
  HandleScope handle_scope;
  int argc = args.Length();

  if (argc < 3)
    return Undefined();

  ostringstream os;
  try {
    if (rapidxml_parse_xml(args, argc, 1, os) < 0)
      return Undefined();
  } catch (parse_error e) {
    printf("rapidxml_parse_xml() : %s when parse the data at %s\n", e.what(), e.where<char>());
    return String::New("",0);
  }
  Handle<String> hString = String::New(os.str().c_str(), os.str().size());

  return handle_scope.Close(hString);
}

static Handle<Value> ku6_copyBetween(const Arguments &args)
{
  HandleScope handle_scope;

  if (args.Length() != 3)
    return Undefined();
  String::AsciiValue input(args[0]->ToString());
  String::AsciiValue start(args[1]->ToString());
  String::AsciiValue end(args[2]->ToString());
  char *ptr = strstr(*input, *start);
  if (ptr) {
    ptr += start.length();
    char *ptr2 = strstr(ptr, *end);
    Handle<String> hString;
    if (ptr2)
      hString = String::New(ptr, ptr2-ptr);
    else
      hString = String::New(ptr);// We assume the whole string is qualified
    return handle_scope.Close(hString);
  }

  return Undefined();
}

//Note: we assume |src| is enough to hold all the replacement of |strNew|
static char* strReplace(char *src, int maxLen, const char *strOld, const char *strNew, int count)
{
  char *pTemp = src;
  int iLength, iDiff;

  if (src == NULL || strOld == NULL || strNew == NULL)
    return NULL;

  iLength = strlen(strOld);
  iDiff = iLength - strlen(strNew);

  if (iLength == 0)
   return src;

  while ((pTemp = strstr(pTemp, strOld)) && count > 0) {
    const char *p = strNew;
    if (iDiff < 0) {
      int i = strlen(src);
      if (i-iDiff >= maxLen)
        return NULL;
      while (i >= pTemp-src+iLength) {
        src[i-iDiff] = src[i];
        i--;
      }
    }
    while (*p != '\0')
      *pTemp++ = *p++;

    //Note: MUST not use strcpy(), because the memory is overlapped
    //the CRT is undefined in such scenario
#if 0
    if(iDiff > 0)
      strcpy(pTemp, pTemp+iDiff);
#else
    if(iDiff > 0)
      memmove(pTemp, pTemp+iDiff,strlen(pTemp+iDiff)+1);
#endif

    count--;
  }

  return src;
}

static Handle<Value> ku6_strReplace(const Arguments &args)
{
  HandleScope handle_scope;
  if (args.Length() != 3)
    return Undefined();

  String::AsciiValue input(args[0]->ToString());
  String::AsciiValue strOld(args[1]->ToString());
  String::AsciiValue strNew(args[2]->ToString());

  if (!*input || !*strOld || !*strNew)
    return Undefined();

  int len = strlen(*input);
  int cache_size = len + PAGE_CACHE_SIZE;
  // Avoid overflow
  if (cache_size < len)
    return Undefined();
  char *cache = new char[cache_size];
  if (!cache)
    return Undefined();
  strcpy(cache, *input);

  char *ptr = strReplace(cache, cache_size, *strOld, *strNew, 0x7FFFFFFF);
  if (ptr) {
    Handle<String> hString = String::New(ptr);
    delete[] cache;
    return handle_scope.Close(hString);
  }
  delete[] cache;

  return Undefined();
}

static Handle<Value> ku6_strFormat(const Arguments &args)
{
  HandleScope handle_scope;
  int i, argc = args.Length();

  if (argc < 2)
    return Undefined();

  String::AsciiValue input(args[0]->ToString());
  if (!*input)
    return Undefined();

  int len = strlen(*input);
  int cache_size = len + PAGE_CACHE_SIZE;
  if (cache_size < len)
    return Undefined();
  char *cache = new char[cache_size];
  if (!cache)
    return Undefined();
  strcpy(cache, *input);

  char key[16], *ptr = cache;
  for (i=1; i<argc; i++) {
    sprintf(key, "%%%d", i);
    String::AsciiValue replace(args[i]->ToString());
    char *ptrMatch = strstr(ptr, key);
    if (ptrMatch) {
      len = cache_size - (ptr - cache);
      if (!strReplace(ptr, len, key, *replace, 0x7FFFFFFF)) {
        delete[] cache;
        return Undefined();
      }
      ptr += strlen(*replace);
    }
  }
  Handle<String> hString = String::New(cache);
  delete[] cache;

  return handle_scope.Close(hString);
}

static int hexchar2int(char c)
{
  if (c >= '0' && c <= '9') return c - '0';
  if (c >= 'a' && c <= 'f') return c - 'a' + 10;
  if (c >= 'A' && c <= 'F') return c - 'A' + 10;
  return -1;
}

static char * decodeURL(char *URL)
{
  if (!URL)
    return NULL;

  char *p = strchr(URL, '%');

  // Replace hex chars to ascii chars
  while (p) {
    char strOld[4];
    char strNew[2];
    unsigned int v = hexchar2int(p[1]);

    v <<= 4;
    v |= hexchar2int(p[2]);

    strNew[0] = v;
    strNew[1] = 0;

    strOld[0] = p[0];
    strOld[1] = p[1];
    strOld[2] = p[2];
    strOld[3] = 0;

    strReplace(p, 4, strOld, strNew, 1);
    // MUST be |p++|, because |strNew| could be '%'
    p++;
    p = strchr(p,'%');
  }

  return URL;
}

static char * cleanURL(char *URL)
{
  char *p0 = strchr(URL, '?');
  char *p = strchr(URL, '%');
  if (p0 && p) {
    // we only modify the unique URL 'http:%3Axxx%2Fyyyy?zzzz'
    // we do not accept any URL with parameters '%' after '?'
    if (p0 < p)
      return URL;
  }

  decodeURL(p);

  return URL;
}

static Handle<Value> ku6_clean_url(const Arguments &args)
{
  HandleScope handle_scope;
  if (args.Length() != 1)
    return Undefined();

  String::AsciiValue input_string(args[0]->ToString());
  if (!*input_string)
    return Undefined();

  char *cache_string = new char[strlen(*input_string)];
  if (!cache_string)
    return Undefined();
  strcpy(cache_string, *input_string);
  cleanURL(cache_string);
  Handle<String> hString = String::New(cache_string);
  delete[] cache_string;

  return handle_scope.Close(hString);
}

static Handle<Value> ku6_alert(const Arguments &args)
{
  HandleScope handle_scope;
  if (args.Length() != 1)
    return Undefined();

  String::AsciiValue msg(args[0]->ToString());
#if defined(_MSC_VER)
  ::MessageBox(NULL, *msg, "JSDownloader",MB_OK);
#else
  printf("%s\n", *msg);
#endif

  return Undefined();
}

static int download_notify_zhuaqu(entity_t *en, int type)
{
  int ret = 0, len = 0;
  char *s = NULL;
  if((s=strstr(en->data,"zhuaqu=1"))){
    len = sizeof(NOTIFY_ZHUAQU_URL);
    len += strlen(en->data) + 1;
    char *url = (char*)thread_calloc(len);
    WebPage page;
    memset(&page, 0, sizeof(WebPage));
    snprintf(url,len,NOTIFY_ZHUAQU_URL,type,en->data);
    printf("download_notify_zhuaqu():url:%s type:%d\n",url,type);
    if(curl_download(curl_write_mem, &page ,url,0,NULL,0,NOTIFY_UTCC_TIMEOUT) < 0){
      printf("download_notify_zhuaqu():Error url: %s\n",url);
      ret = -1;
    }else{
      printf("download_notify_zhuaqu():successful result: %s\n",(char*)page.data);
    }
    thread_free(page.data,page.size+page.left);
    thread_free(url,len);
  }
  return ret;
}

static int download_notify_zhuantie(entity_t *en, int type)
{
  int ret = 0, len = 0;
  char *s = NULL;
  if((s=strstr(en->data,"sid="))){
    switch (type) {
    case FAILURE_NOTIFY:
      len = sizeof(NOTIFY_ZHUANTIE_FAILURE_URL);
      break;
    case SUCCESS_NOTIFY:
      len = sizeof(NOTIFY_ZHUANTIE_SUCCESS_URL);
      break;
    default:
      return -1;
    }
    len += strlen(en->data) + 1;
    char *url = (char*)thread_calloc(len);
    WebPage page;
    memset(&page, 0, sizeof(WebPage));
    switch (type) {
    case FAILURE_NOTIFY:
      sprintf(url, NOTIFY_ZHUANTIE_FAILURE_URL, en->data);
      break;
    case SUCCESS_NOTIFY:
      sprintf(url, NOTIFY_ZHUANTIE_SUCCESS_URL, en->data);
      break;
    default:
      return -1;
    }
    printf("download_notify_zhuantie():data:%s type:%d\n",url,type);
    if(curl_download(curl_write_mem, &page ,url,0,NULL,0,NOTIFY_UTCC_TIMEOUT) < 0){
      printf("download_notify_zhuantie():Error data: %s\n",en->data);
      ret = -1;
    }else{
      printf("download_notify_zhuantie():successful result: %s\n",(char*)page.data);
    }
    thread_free(page.data,page.size+page.left);
    thread_free(url,len);
  }
  return ret;
}

static int download_notify_utcc(entity_t *en)
{
  int ret = 0;
  //int request_data_len = strlen(en->data);
  if(strstr(en->data,"url=") && strstr(en->data,"sid=")){
    WebPage page;
    memset(&page, 0, sizeof(WebPage));
    //sprintf(en->data+request_data_len,"%s",final_file+strlen(en->path));

    printf("notify_utcc_url %s \nReport data: %s\n",notify_utcc_url,en->data);
    if(curl_download(curl_write_mem, &page ,notify_utcc_url, 1, en->data, 0,NOTIFY_UTCC_TIMEOUT) < 0){
      printf("Report Error data: %s\n",en->url);
      ret = -1;
    }else{
      printf("Report successful result: %s\n",(char*)page.data);
      if(strstr(page.data,"err")){
        ret = -1;
      }
    }
    thread_free(page.data,page.size+page.left);
    if(ret == 0)
      download_notify_zhuantie(en, SUCCESS_NOTIFY);
  }
  return ret;
}

static int download_notify(entity_t *en,char *final_file,int type)
{
  int ret = 0;
  if(type == SUCCESS_NOTIFY){
	  int request_data_len = strlen(en->data);
	  sprintf(en->data+request_data_len,"%s",final_file+strlen(en->path));
	  if(strstr(en->data,"zhuaqu=1")){
		  ret = download_notify_zhuaqu(en,ZHUAQU_SUCCESS_NOTIFY);
	  }else{
		  ret = download_notify_utcc(en);
	  }
  }else{
	  if(strstr(en->data,"zhuaqu=1")){
		  ret = download_notify_zhuaqu(en,type);
	  }else{
		  ret = download_notify_zhuantie(en,FAILURE_NOTIFY);
	  }
  }
  return ret;
}

static int timed_system(const char *cmd,int seconds)
{
  int stat;
  pid_t pid;
  struct sigaction sa, savintr, savequit;
  sigset_t saveblock;
  if (cmd == NULL)
      return(1);
  sa.sa_handler = SIG_IGN;
  sigemptyset(&sa.sa_mask);
  sa.sa_flags = 0;

  sigemptyset(&savintr.sa_mask);
  sigemptyset(&savequit.sa_mask);

  sigaction(SIGINT, &sa, &savintr);
  sigaction(SIGQUIT, &sa, &savequit);

  sigaddset(&sa.sa_mask, SIGCHLD);
  sigprocmask(SIG_BLOCK, &sa.sa_mask, &saveblock);
  if ((pid = fork()) == 0) {
      sigaction(SIGINT, &savintr, (struct sigaction *)0);
      sigaction(SIGQUIT, &savequit, (struct sigaction *)0);
      
      sigprocmask(SIG_SETMASK, &saveblock, (sigset_t *)0);
      alarm(seconds);
      execl("/bin/sh", "sh", "-c", cmd, (char *)0);
      _exit(127);
  }
  if (pid == -1) {
      stat = -1; /* errno comes from fork() */
  } else {
      while (waitpid(pid, &stat, 0) == -1) {
          if (errno != EINTR){
              stat = -1;
              break;
          }
      }
      printf("WIFEXITED(%d) WEXITSTATUS(%d) WIFSIGNALED(%d) WTERMSIG(%d)\n",
             WIFEXITED(stat),WEXITSTATUS(stat),WIFSIGNALED(stat),WTERMSIG(stat));

  }
  sigaction(SIGINT, &savintr, (struct sigaction *)0);
  sigaction(SIGQUIT, &savequit, (struct sigaction *)0);
  sigprocmask(SIG_SETMASK, &saveblock, (sigset_t *)0);
  return(stat);
}

static int ku6_download_files(char *pURL, char sep, entity_t *en)
{
  char *output_path = en->path;
  char *filename, *p;
  int file_count = 1, ret;

  pool_t   *pool =  thread_mem_pool();
  if(!pool) return -1;

  // Avoid NULL and zero length string.
  if (!pURL || !*pURL) {
    printf("ku6_download_files():the media url returned by JS is NULL\n");
    return -1;
  }

  int success_files_init_len = 512;
  int success_files_len = success_files_init_len;
  int file_path_max_len = 512;
  char *successful_files  = (char*)thread_calloc(success_files_len);
  if(!successful_files) return -1;
  sprintf(successful_files,"/data/ku6downloader/lib/bin/fastreaming ");

  filename = (char*)thread_calloc(file_path_max_len);
  if(!filename) {
    //thread_free(successful_files, success_files_len);
    return -1;
  }

  int is_mms = !strncmp(pURL,"mms://", 6);
  do {
    p = strchr(pURL, sep);
    if (p)
      *p = '\0';
    if (is_mms) {
      //because curl does not support mms:// protocol, we try to use |fastreaming| to download
      sprintf(filename, "/data/ku6downloader/lib/bin/fastreaming -i \"%s\" -o %s_file_%04d.wmv", pURL, output_path, file_count++);
      if(timed_system(filename,DOWNLOAD_MMS_FILE_TIMEOUT) != 0){
      //if(timed_system(filename,3) != 0){
      	if(p)
          *p = sep;
        printf("download mms:%s error:%s\n",filename,strerror(errno));
      	return -1;
      }
    } else {
#if defined(USE_CURL_API_DOWNLOAD)
      sprintf(filename, "%s_file_%04d.flv", output_path, file_count++);
      printf("ku6_download_files(): filename:%s\n",filename);
      FILE *fp = fopen(filename, "wb");
      if (!fp) {
        printf("the file %s can`t be opened! error(%u):%s\n",output_path,errno,strerror(errno));
        if (p)
          *p = sep;
        return -1;
      }
      if (curl_download(curl_write_file, fp, pURL, 0, NULL, 1,DOWNLOAD_FILE_TIMEOUT) < 0) {
        if(p)
          *p = sep;
        fclose(fp);
        return -1;
      }
      fclose(fp);
#else
      sprintf(filename, "curl --compressed --max-redirs %d -A \"%s\" -L -o %s_file_%04d.flv \"%s\"", MAX_REDIRECT_NUM, g_user_agent, output_path, file_count++, pURL);
      timed_system(filename,DOWNLOAD_FILE_TIMEOUT);
#endif
    }
    
    int cur_len = (strlen(successful_files) + strlen(filename) + sizeof(INPUT_FILE_ARG_FORMAT));
    char *tmp = (char*)thread_realloc(successful_files,success_files_len,cur_len);
    if(!tmp){
    	printf("thread_realloc successful_files error oldlen:%d newlen:%d\n",success_files_len,cur_len);
    	return -1;
    }
    successful_files = tmp;
    if(cur_len > success_files_len)
      success_files_len = cur_len;
    sprintf(successful_files + strlen(successful_files),INPUT_FILE_ARG_FORMAT"%s",filename);
    if (p)
      *p++ = sep;
    pURL = p;
  } while(pURL);

  if(file_count == 2){
    ret = download_notify(en,filename,SUCCESS_NOTIFY);
    return ret;
  }else{
    if (is_mms)
      sprintf(filename, "%s_file_final.wmv", output_path);
    else if(en->finalname[0]){
      char *lastp = strrchr(output_path,'/');
      lastp++;
      *lastp = '\0';
      sprintf(filename, "%s%s.mp4", output_path,en->finalname);
    }else
      sprintf(filename, "%s_file_final.mp4", output_path);

    printf("final filename xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx:%s",filename);

    int cur_len = (strlen(successful_files) + strlen(filename) + sizeof(OUTPUT_FILE_ARG_FORMAT) + 1);
    char *tmp = (char*)thread_realloc(successful_files,success_files_len,cur_len);
    if(!tmp){
    	printf("thread_realloc successful_files error oldlen:%d newlen:%d\n",success_files_len,cur_len);
    	return -1;
    }
    successful_files = tmp;
    if(cur_len > success_files_len)
       success_files_len = cur_len;
    sprintf(successful_files + strlen(successful_files),OUTPUT_FILE_ARG_FORMAT"%s",filename);

    printf("ku6_download_files():successful_files %s\n",successful_files);
    if(timed_system(successful_files,MERGE_FILE_TIMEOUT) == 0){
      printf("ku6_download_files():combine successful_files filename:%s\n",filename);
      ret = download_notify(en,filename,SUCCESS_NOTIFY);
      return ret;
    }else{
      printf("ku6_download_files():combine file failure, try to use the first file\n");
      char *first_file_s = strstr(successful_files,"-i");
      if(first_file_s){
        first_file_s = first_file_s + 2;
        while(*first_file_s == ' ')first_file_s++;
        char *first_file_e = strchr(first_file_s,' ');
        if(first_file_e) *first_file_e = '\0';
        ret = download_notify(en,first_file_s,SUCCESS_NOTIFY);//just notify the first file
        if(first_file_e) *first_file_e = ' ';
        return ret;
      }
    }
  }
  //thread_free(successful_files, success_files_len > success_files_init_len ? success_files_len : success_files_init_len);
  //thread_free(filename, file_path_max_len);
  return 0;
}

Handle<Value> execute_js_function(const char *jsScript,const char * js_fun_name,const char * js_fun_param)
{
  //Handle<Context> context = Context::GetCurrent();
  Handle<Context> context = Context::GetEntered();

  // Create a string containing the JavaScript source code.
  int js_len =  strlen(jsScript);
  Handle<String> source = String::New(jsScript,js_len);

  // Compile the source code.
  Handle<Script> script = Script::Compile(source);

  // Run the script to get the result.
  Handle<Value> result = script->Run();

  Handle<String> js_func_name = String::New(js_fun_name);
  Handle<Value>  js_func_val = context->Global()->Get(js_func_name);

  if(js_func_val->IsFunction()){
    Handle<Function> js_fun = Handle<Function>::Cast(js_func_val);
    Handle<Value>  js_param = String::New(js_fun_param);
    result = js_fun->Call(context->Global(),1,&js_param);
  }

  return result;
}

static char* chooseJSData(char *url)
{
  if(!url) return NULL;
  printf("chooseJSData: %s\n",url);
  char *site = strchr(url,'.');
  if(!site) return NULL;

  char *e = strpbrk(site,"/:?");
  if(!e)  e = (char*)url + strlen(url);
  char oldc = *e;
  *e = '\0';
  hash_iter_t iter;
  hash_iter_init(JSData,&iter);
  hash_elt_t *elt = NULL;
  while((elt=hash_iter_next(&iter)) != NULL){
    char *name = elt->name;
    if(strstr(site,name)){
      *e = oldc;
      printf("chooseJSData(): site name:%s\n",name);
      return name;
    }
  }

  *e = oldc;
  return NULL;
}

static void cleanJSData(){
  hash_iter_t iter;
  hash_iter_init(JSData,&iter);
  hash_elt_t *elt = NULL;
  while((elt=hash_iter_next(&iter)) != NULL){
    char *name = elt->name;
    void *data = elt->data;
    printf("cleanJSData():recycle name:%s(%ld) data:%p(%ld)\n",name,strlen(name),(char*)data,strlen((char*)data));
    thread_recycle(name,strlen(name)+1);
    thread_recycle(data,strlen((char*)data)+1);
  }
}

static void  makeup_dir(char *path)
{
  if(!path || !strlen(path)) return ;
  int ret = 0;
  char *p = strchr(path,'/');
  if(!p) return;
  char *l;
  while((l=strchr(++p,'/'))){
    *l = '\0';
    printf("makeup_dir():path:%s\n",path);
    ret = mkdir(path,S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH);
    *l = '/';
    p = l;
  }
}

static void reset_entity(entity_t *en)
{
  if(!en) return ;
  memset(en->data,0,ENTITY_MAX_POST_DATA_BUFF+1);
  memset(en->url,0,ENTITY_MAX_URL_BUFF+1);
  memset(en->domain,0,ENTITY_MAX_DOMAIN_BUFF+1);
  memset(en->path,0,ENTITY_MAX_OUTPATH_BUFF+1);
  memset(en->finalname,0,ENTITY_MAX_OUTPATH_BUFF+1);
}

static int makeup_entity(entity_t *en, char *output, char *finalname)
{
  if(!en || !strlen(en->data)){
    if(en)
      printf("makeup_entity():get request data from nginx is empty!\n");
    return -1;
  }

  char *url = strstr(en->data,"url=");
  if(url){
    url = url + 4;
  }else{
    url = en->data;
  }
  char *url_end = strchr(url,'&');
  if(url_end){
    *url_end = '\0';
  }
  memcpy(en->url,url,ENTITY_MAX_URL_BUFF);

  if(url_end){
    *url_end = '&';
  }
  cleanURL(en->url);

  //for domain
  char *domain = chooseJSData(en->url);
  if(!domain){
    printf("makeup_entity():unsupported web site:%s\n",en->url);
    return -1;
  }
  printf("makeup_entity():chooseJSData %s\n",domain);
  memcpy(en->domain,domain,ENTITY_MAX_DOMAIN_BUFF);

  if(finalname && finalname[0]){
     //snprintf(en->path,ENTITY_MAX_OUTPATH_BUFF,"%s%s/",output,en->domain);
     snprintf(en->finalname,ENTITY_MAX_OUTPATH_BUFF,"%s",finalname);
  }
     //for en->data
     /*strcat(en->data,"&path=/");
     strcat(en->data,en->domain);
     strcat(en->data,"/");*/
  //}else{

    //for output_path
    uuid_t uuid;
    //uuid_generate_time(uuid);
    uuid_generate(uuid);
    //uuid_generate_random(uuid);
    char uuid_str[17] = {'\0'};
    uuid_unparse(uuid,uuid_str);
    snprintf(en->path,ENTITY_MAX_OUTPATH_BUFF,"%s%s/%s",output,en->domain,uuid_str);
    printf("makeup_entity():entity->path  %s\n",en->path);
    makeup_dir(en->path);

    //for en->data
    strcat(en->data,"&path=/");
    strcat(en->data,en->domain);
    strcat(en->data,"/");
    strcat(en->data,uuid_str);

  //}



  printf("makeup_entity():end en->url:%s en->domain:%s en->path:%s en->data:%s\n",en->url,en->domain,en->path,en->data);
  return 0;
}

Handle<ObjectTemplate> createGlobalFunc()
{
  // Create global C++ register template.
  Handle<ObjectTemplate> global_templ = ObjectTemplate::New();

  // Register C++ function to V8.
  //global_templ->Set(String::New("character_convert"), FunctionTemplate::New(character_convert));
  global_templ->Set(String::New("download_page"), FunctionTemplate::New(ku6_download_page));
  global_templ->Set(String::New("download_page_post"), FunctionTemplate::New(ku6_download_page_post));
  global_templ->Set(String::New("parse_xml"), FunctionTemplate::New(ku6_parse_xml));
  global_templ->Set(String::New("parse_xml_attribute"), FunctionTemplate::New(ku6_parse_xml_attribute));
  global_templ->Set(String::New("copyBetween"), FunctionTemplate::New(ku6_copyBetween));
  global_templ->Set(String::New("strReplace"), FunctionTemplate::New(ku6_strReplace));
  global_templ->Set(String::New("strFormat"), FunctionTemplate::New(ku6_strFormat));
  global_templ->Set(String::New("clean_url"), FunctionTemplate::New(ku6_clean_url));
  global_templ->Set(String::New("alert"), FunctionTemplate::New(ku6_alert));

  return global_templ;
}

char * validateVideoUrl(char *url)
{
  // Create a stack-allocated handle scope.
  HandleScope handle_scope;

  Handle<ObjectTemplate> global_templ = createGlobalFunc();
  // Create a new context.
  Persistent<Context> context = Context::New(NULL, global_templ);

  // Enter the created context for compiling.
  Context::Scope context_scope(context);

  char *domain = chooseJSData(url);
  if(!domain){
    printf("validateVideoUrl():error:invalidate url\n");
    return NULL;
  }
  char *pJSData = (char*)hash_find(JSData,domain);
  if(!pJSData){
     printf("hash_find():error:have`nt the JSData of domain(%s)\n",domain);
     return NULL;
  }
  printf("validateVideoUrl():download the urls:%s  domain:%s  \n",url,domain);
  //thread_recycle(domain,strlen(domain)+1);
  Handle<Value> ret =  execute_js_function(pJSData,"validateVideoUrl",url);
  String::AsciiValue ascii(ret);
  printf("%s\n",*ascii);

  context.Dispose();
  return *ascii;
}

int do_download(entity_t *en)
{
  if(!en) return -1;

  pthread_rwlock_rdlock(&JSData_RWlock);
  char *pJSData = (char*)hash_find(JSData,en->domain);
  if(!pJSData){
    printf("do_download():have`nt the JSData of domain(%s)\n",en->domain);
    pthread_rwlock_unlock(&JSData_RWlock);
    return -1;
  }

  HandleScope handle_scope;

  //validate the duration limited
  char *stime=NULL , *etime=NULL;
  if((stime=strstr(en->data,"stime=")) && (etime=strstr(en->data,"etime="))){
   char *flag = strchr(stime,'&');
   if(flag) *flag = '\0';
   int st = atoi(stime+6);
   if(flag) *flag = '&';

   flag = strchr(etime,'&');
   if(flag) *flag = '\0';
   int et = atoi(etime+6);
   if(flag) *flag = '&';
   printf("validateVideoUrl():download the urls:%s  domain:%s stime:%d  etime:%d \n",en->url,en->domain,st,et);
   Handle<Value> ret =  execute_js_function(pJSData,"validateVideoUrl",en->url);
   String::AsciiValue ascii(ret);
   printf("validateVideoUrl() result:download the urls:%s  domain:%s stime:%d  etime:%d result:%s\n",en->url,en->domain,st,et,*ascii);
   if(!*ascii) return -1;
   char *seconds = NULL;
   if((seconds=strstr(*ascii,"seconds:"))){
	   int sec = atoi(seconds+8);
	   if(sec <st || sec > et ) //over the duration
		   return OVERDURA_FAILURE_NOTIFY;
   }else if(!strstr(*ascii,"success_title:")){
	   return -1;
   }
  }
  printf("do_download():download the urls:%s  domain:%s  \n en->url:%s\n",en->url,en->domain,en->url);
  Handle<Value> ret =  execute_js_function(pJSData,"js_get_media_url",en->url);
  String::AsciiValue ascii(ret);
  printf("do_download():get flv paths:%s\n",*ascii);
  /*char ascii [200];
  snprintf(ascii,200,"%s","http://f.youku.com/player/getFlvPath/st/flv/fileid/03000201004F31B1637B61006CB892F64C08F3-CF06-CCB0-DC29-A5BB71318C61?K=6a03f3aee151e316261c2898&hd=0");*/
  pthread_rwlock_unlock(&JSData_RWlock);

  // Extract muli-URLs which is seperated by SEP_INDICATOR_CHAR.
  return ku6_download_files(*ascii, SEP_INDICATOR_CHAR,en);
}


static uint32_t* ComputeStackLimit(uint32_t size) {
  uint32_t* answer = &size - (size / sizeof(size));
  // If the size is very large and the stack is very near the bottom of
  // memory then the calculation above may wrap around and give an address
  // that is above the (downwards-growing) stack.  In that case we return
  // a very low address.
  if (answer > &size) return reinterpret_cast<uint32_t*>(sizeof(size));
  return answer;
}

void * download_worker(void *_idx)
{
  printf("download_worker():create %ld worker success! and the global block_queue size:%d:%d\n",pthread_self(),worker_entities.size,*((int*)_idx));
  int ret  = 0;
  if((ret = thread_setspecific(_idx)) != 0){
    printf("%ld thread_setspecific() error:%s\n",pthread_self(),strerror(ret));
    return (void*)0;
  }

  void * retval = thread_calloc(sizeof(int));
  memcpy(retval,_idx,sizeof(int));

  // Set heap limits.

  /*uint32_t* stack_limit = ComputeStackLimit(1024 * 1024 * 2);

  v8::ResourceConstraints constraints;
  constraints.set_max_young_space_size(1024 * 1024);
  constraints.set_max_old_space_size(4 * 1024 * 1024);
  // Set stack limit.
  constraints.set_stack_limit(stack_limit);
  //v8::SetResourceConstraints(&constraints);*/

  //for multithread
  Isolate *isolate = Isolate::New();
  Isolate::Scope iscope(isolate);
  // Create a stack-allocated handle scope.
  HandleScope handle_scope;

  Handle<ObjectTemplate> global_templ = createGlobalFunc();
  // Create a new context.
  Persistent<Context> context = Context::New(NULL, global_templ);

  // Enter the created context for compiling.
  Context::Scope context_scope(context);

  //V8::AdjustAmountOfExternalAllocatedMemory(1024 * 1024 * 10);

  for(;;){
    queue_t *qq  = block_queue_pop(&worker_entities);
    if(qq == NULL && worker_entities.block_flag == 0) break; // thread exit;
    entity_t *en = queue_data(qq,entity_t,queue);
    int ret = 0;
    if((ret=do_download(en)) != 0)
      download_notify(en,NULL,ret);
    thread_reset_pool();
    reset_entity(en);
    block_queue_push(&free_entities,&en->queue);
    pool_print(thread_mem_pool()); 
    long t1 = time(NULL);
    V8::LowMemoryNotification(); 
    long t2 = time(NULL);
    printf("LowMemoryNotification escape:%ld\n",(t2-t1));
  }
  //thread_destroy_pools();
  context.Dispose();
  //isolate->Dispose();
  return retval;
}

void free_entity(void *p)
{
  if(!p) return;
  queue_t *q = (queue_t*)p;
  entity_t *en = queue_data(q,entity_t,queue);
  printf("free_entity():%p\n",en);
  free(en);
}

entity_t *create_new_entity()
{
  entity_t *en = (entity_t*)thread_calloc(sizeof(entity_t));
  if(!en) return NULL;
  printf("create_new_entity():%p:%p\n",&en->queue,en);
  return en;
}

void  destoryAll()
{
  destroy_pools();
  //block_queue_destory(&free_entities,free_entity);
  //block_queue_destory(&worker_entities,free_entity);
  hash_destory(JSData);
  pthread_rwlock_destroy(&JSData_RWlock);
}

static void load_worker_file(char *output)
{
  DIR *dp;
  struct dirent *dir;
  FILE *fp;
  char fpath[MAX_PATH_LEN+1]={'\0'};

  if((dp=opendir(QUEUE_MOREDATA_PATH)) == NULL)
     return;

  while((dir=readdir(dp)) && RunCheckFileThread){
    if(strcmp(dir->d_name,".") == 0 || strcmp(dir->d_name,"..")== 0)
      continue;
    snprintf(fpath,sizeof(fpath),"%s/%s",QUEUE_MOREDATA_PATH,dir->d_name);
    printf("read (%s)\n",fpath);
    if((fp=fopen(fpath,"r")) == NULL){
      printf("fopen (%s) error:%s\n",fpath,strerror(errno));
    }else{
      entity_t *en = NULL;
      for(;RunCheckFileThread;){
        if(!en){
          queue_t *qq = block_queue_pop(&free_entities);
          printf("main(%ld) thread pop free_entities done\n",pthread_self());
          if(qq)
            en = queue_data(qq,entity_t,queue);
          else
            en = create_new_entity();
          if(!en){
            printf("get free entity error!\n");
            destoryAll();
            exit(1);
          }
        }
        if((fgets(en->data,ENTITY_MAX_POST_DATA_BUFF,fp)) != NULL){
          char *tmp = strchr(en->data,'\n');
          if(tmp) *tmp = '\0';
          if(makeup_entity(en,output,NULL) != 0){
            printf("makeup_entity error!\n");
            continue;
          }
          block_queue_push(&worker_entities,&en->queue);
          printf("main thread push entity.queue:%p  url:%s  free_entities.size:%d\n",&en->queue,en->url,free_entities.size);
          en = NULL;
        }else{
          fclose(fp);
          unlink(fpath);
          break;
        }
      }
    }
  }

  closedir(dp);
}

void* check_file_worker(void *arg){
	char *output_path = (char*)arg;
	printf("check_file_worker begin to while for chect output_path:%s\n",output_path);
	while(RunCheckFileThread){
		load_worker_file(output_path);
		printf("check_file_worker goto sleep(10s) for check output_path:%s\n",output_path);
		sleep(10);
	}
	printf("check_file_worker exit to while for chect output_path:%s\n",output_path);
	return arg;
}

/*void init_check_file_worker_thread(char *out_path){
  int ret;
  pthread_t tid;
  pthread_attr_t attr;
  ret = thread_key_create(free);
  if(ret != 0){
	printf("thread_key_create():error: %s\n",strerror(ret));
	destoryAll();
	exit(ret);
  }
  ret = pthread_attr_init(&attr);
  ret = pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);
  ret = pthread_create(tid,NULL,check_file_worker,(void*)idx);
  if(ret != 0){
	 printf("init_check_file_worker_thread():cant create thread %s\n",strerror(ret));
	 destoryAll();
	 exit(ret);
  }
  pthread_attr_destroy(&attr);
}*/

void init_threads(int threadcnt,char *out_path)
{
  int i,ret;
  pthread_attr_t attr;
  ret = thread_key_create(free);
  if(ret != 0){
    printf("thread_key_create():error: %s\n",strerror(ret));
    destoryAll();
    exit(ret);
  }
  ret = pthread_attr_init(&attr);
  ret = pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);
  threads = (pthread_t**)thread_calloc(sizeof(pthread_t*)*threadcnt);
  for(i=0;i<threadcnt;i++){
    pthread_t *tid = (pthread_t*)thread_calloc(sizeof(pthread_t));
    threads[i] = tid;
    if(i==(threadcnt-1)){//last one
		char *path = (char*)malloc(ENTITY_MAX_OUTPATH_BUFF); // free by thread_exit auto
		strcpy(path,out_path);
		ret = pthread_create(tid,NULL,check_file_worker,(void*)path);
    }else{
		int *idx = (int*)malloc(sizeof(int));
		*idx = i+1;
		ret = pthread_create(tid,NULL,download_worker,(void*)idx);
    }
    if(ret != 0){
	  printf("pthread_create():cant create worker %s\n",strerror(ret));
	  destoryAll();
	  exit(ret);
	}
  }
  pthread_attr_destroy(&attr);
}



static void * readFile(const char *filename)
{
  FILE *fp = fopen(filename, "r");
  if (!fp) {
    printf("Cann't open file: %s \n",filename);
    return NULL;
  }
  fseek(fp, 0, SEEK_END);
  int len = ftell(fp);
  fseek(fp, 0, SEEK_SET);
  // MUST be +1
  // NOTE: "\r\n"(0x0d0x10) -->"\n"(0x10) will reduce file size,
  // We MUST zero the mem, terminate string with NULL.
  void *ret = thread_calloc(len+1);
  if (!ret) {
    printf("thread_calloc():memory allocation fails\n");
    fclose(fp);
    return NULL;
  }

  fread(ret, len, 1, fp);
  fclose(fp);

  return ret;
}

static void initJS(char *js)
{
  JSData = hash_init(10);
  DIR *dp;
  struct dirent *dirp = NULL;
  struct stat statbuf;

  if(lstat(js,&statbuf) < 0){
    printf("initJS():lstat file %s error\n",js);
    destoryAll();
    exit(1);
  }

  if(S_ISDIR(statbuf.st_mode) == 0){
    if(!strstr(js,".js")){
      printf("initJS():pleases input js file which is suffixed by \".js\"\n");
      destoryAll();
      exit(1);
    }
    char *prefix = strchr(js,'.');
    int  prefix_len = prefix - dirp->d_name;
    char *name = (char*)thread_calloc(prefix_len+1);
    if(!name){
      destoryAll();
      exit(1);
    }
    memcpy(name,js,prefix_len);
    char *p = name;
    while(*p){
      *p = tolower(*p);
      p++;
    }

    void *tmp = readFile(js);
    hash_push(JSData,name,tmp);
  }else{
    char *end = js + strlen(js);
    if((dp=opendir(js)) == NULL){
      printf("initJS():Can`t read path %s\n",js);
      destoryAll();
      exit(1);
    }

    while((dirp=readdir(dp)) != NULL){
      if(strcmp(dirp->d_name,".") == 0 ||
        strcmp(dirp->d_name,"..") == 0)
        continue;
      if(!strstr(dirp->d_name,".js"))
        continue;
      char *prefix = strchr(dirp->d_name,'.');
      int  prefix_len = prefix - dirp->d_name;
      char *name = (char*)thread_calloc(prefix_len+1);
      if(!name){
        destoryAll();
        exit(1);
      }
      memcpy(name,dirp->d_name,prefix_len);
      char *p = name;
      while(*p){
        *p = (char)tolower(*p);
        p++;
      }
      strcat(end,"/");
      strcat(end+1,dirp->d_name);
      printf("initJS():load js file :%s name:%s url:%s\n",dirp->d_name,name,js);
      void *tmp = readFile(js);
      hash_push(JSData,name,tmp);
      *(end)  = '\0';
    }
    closedir(dp);
  }
  printf("initJS():JSData count:%d\n",JSData->count);
}



void ReeadJSData(int signo){
  pthread_rwlock_wrlock(&JSData_RWlock);
  cleanJSData();
  hash_destory(JSData);
  initJS(input_js);
  pthread_rwlock_unlock(&JSData_RWlock);
}

void GracefulTerminate(int signo){
  //stop check_file_worker thread
  RunCheckFileThread = 0;

  //wait for finish current workers;
  while(worker_entities.size >0 ){
    printf("wait worker_entities.size(%d) to zero\n",worker_entities.size);
    usleep(1000*1000);
  }

  //wake_up threads
  broadcast_threads_wakeup(&worker_entities,0);

  //get per thread exit data;
  int threadcnt = MAX_DISPATHER_THREADS+1;
  int err = 0;
  while(threadcnt>0){
    void *idx;
    if((err=pthread_join(*threads[threadcnt-1],&idx)) != 0){
      printf("can`t join thread(%d) error:%s\n",threadcnt,strerror(err));
    }else{
      printf("%d thread ret:%d\n",threadcnt,*((int*)idx));
    }
    broadcast_threads_wakeup(&worker_entities,0);
    threadcnt--;
  }

  //exit
  destoryAll();
  exit(0);
}

int main(int argc, char* argv[])
{
  int i;
  // MUST initialize
  char input_url[ENTITY_MAX_URL_BUFF+1]         = {'\0'};
  char output_path[ENTITY_MAX_OUTPATH_BUFF+1]   = {'\0'};
  char final_name[ENTITY_MAX_OUTPATH_BUFF+1]    = {'\0'};
  strncpy(notify_utcc_url,NOTIFY_SUCCESS_VSYS_URL,NOTIFY_UTCC_URL);

  if(pthread_rwlock_init(&JSData_RWlock,NULL) != 0){
    printf("pthread_rwlock_init  %s error\n",strerror(errno));
    destoryAll();
    exit(1);
  }

  if(signal(SIGHUP,ReeadJSData)==SIG_ERR || signal(SIGTERM,GracefulTerminate)==SIG_ERR){
    printf("signal(SIGHUP|SIGTERM) error:%s\n",strerror(errno));
    destoryAll();
    exit(1);
  }

  thread_init_pools(MAX_DISPATHER_THREADS+1, 300*1024);

  for (i = 1; i < argc-1; i += 2) {
    char *p = argv[i];
    if (*p != '-') {
      printf("error:Argument must be started by character: \'-\'\n");
      destoryAll();
      exit(1);
    }
    p++;
    switch(*p) {
    case 'i':
    case 'I':
      strncpy(input_js, argv[i+1], JS_MAX_PATH_BUFF);
      initJS(input_js);
      break;
    case 'r':
    case 'R':
      strncpy(input_url, argv[i+1], ENTITY_MAX_URL_BUFF);
      break;
    case 'o':
    case 'O':
      strncpy(output_path, argv[i+1], ENTITY_MAX_OUTPATH_BUFF);
      // Append the path endline
      p = output_path + strlen(output_path) - 1;
      if (*p != '/' && *p != '\\') {
        p[1] = '/';
        p[2] = '\0';
        printf("output_path :%s\n",output_path);
      }
      break;
    case 'f':
    case 'F':
      strncpy(final_name, argv[i+1], ENTITY_MAX_OUTPATH_BUFF);
      printf("final_name :%s\n",final_name);
      break;
    case 'n':
    case 'N':
      strncpy(notify_utcc_url, argv[i+1], NOTIFY_UTCC_URL);
      break;
    default:
      printf("Unknown argument: %s\n", p);
      destoryAll();
      exit(1);
      break;
    }
  }

  if(strlen(input_url) == 0 && strlen(output_path) == 0){
    printf("error:input_url and output_path can not be empty at the same time!\n");
    destoryAll();
    exit(1);
  }

  // Invoke exactly once for each application that uses libcurl
  // and before any call of other libcurl functions.
  CURLcode ret = curl_global_init(CURL_GLOBAL_ALL);
  if (ret != CURLE_OK) {
    printf("curl_global_init() fails\n");
    destoryAll();
    exit(1);
  }

  // validate the input_url is a video url.
  if(strlen(input_url) && (strlen(output_path) == 0)){
    validateVideoUrl(input_url);
    destoryAll();
    curl_global_cleanup();
    return 0;
  }

  if(strlen(input_url) && (strlen(output_path))){
    entity_t *en = create_new_entity();
    memcpy(en->data,input_url,strlen(input_url));
    if(makeup_entity(en,output_path,final_name) != 0){
      printf("makeup_entity error!\n");
      curl_global_cleanup();
      return 1;
    }
    Isolate::Scope iscope();
    // Create a stack-allocated handle scope.
    HandleScope handle_scope;

    Handle<ObjectTemplate> global_templ = createGlobalFunc();
    // Create a new context.
    Persistent<Context> context = Context::New(NULL, global_templ);

    // Enter the created context for compiling.
    Context::Scope context_scope(context);
    do_download(en);
    context.Dispose();
    destoryAll();
    curl_global_cleanup();
    return 0;
  }

  CircularQueueInShm* cq=NULL;
  if (init_cq(&cq,200,(char*)QUEUE_MOREDATA_PATH)){
    printf("init nginx shared mem error. ask shilin\n");
    curl_global_cleanup();
    destoryAll();
    return 0;
  }

  block_queue_init(&free_entities,0,0);
  for(i=0;i<MAX_DISPATHER_THREADS;i++){
    entity_t *en = create_new_entity();
    block_queue_push(&free_entities,&en->queue);
  }
  block_queue_init(&worker_entities,MAX_DISPATHER_THREADS,1);

  init_threads(MAX_DISPATHER_THREADS+1,output_path);

  // MUST load worker files after |free_entities| and |worker_entities| have been initialized
  //load_worker_file(output_path);  init by above init_threads method

  entity_t *en = NULL;

  //int flag = 0;
  for(;;){
    if(!en){
      queue_t *qq = block_queue_pop(&free_entities);
      if(qq)
        en = queue_data(qq,entity_t,queue);
      else
        en = create_new_entity();
      if(!en){
        printf("get free entity error!\n");
        destoryAll();
        delete_cq(cq);
        curl_global_cleanup();
        exit(1);
      }
    }
    if (readurl_cq(cq, en->data) < 0){
    //if(!flag){
      //strcpy(en->data,"url=http://v.youku.com/v_playlist/f17013245o1p0.html");
      usleep(1000*500);
      //sleep(5);
      //flag =  1;
      continue;
    }else{
      //flag =  0;
      printf("get  data=%s\n",en->data);
      if(makeup_entity(en,output_path,final_name) != 0){
        printf("makeup_entity error!\n");
        continue;
      }
      block_queue_push(&worker_entities,&en->queue);
      printf("main thread push entity.queue:%p  url:%s  free_entities.size:%d\n",&en->queue,en->url,free_entities.size);
      en = NULL;
    }
  }

  destoryAll();

  delete_cq(cq);

  // Invoke exactly once for each application that uses libcurl.
  curl_global_cleanup();

  return 0;
}
